package com.example.rocketmq.base.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.Charset;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @DESCRIPTION: 发布同步消息
 * @USER: shg
 * @DATE: 2023/3/6 23:57
 */
public class Producer {


    public static void main(String[] args) throws Exception {

        // sendSyncMessage();
        // sendAsyncMessage();
        sendOneWayMessage();
    }

    private static void sendOneWayMessage() throws Exception{
        // 1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        // 2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.11.129:9876;192.168.11.130:9876");
        // producer.setNamesrvAddr("192.168.2.50:9876;192.168.2.51:9876");
        // 3.启动producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            Message message = new Message("base", "Tag3", ("Hello RocketMQ OneWayMessage" + (i + 1)).getBytes(Charset.defaultCharset()));
            // 5.发送消息
            producer.sendOneway(message);
            // 线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        // 6.关闭生产者producer
        producer.shutdown();
    }

    // 发送单向消息

    // 发送异步消息
    private static void sendAsyncMessage() throws Exception {
        // 1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        // 2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.11.129:9876;192.168.11.130:9876");
        // producer.setNamesrvAddr("192.168.2.50:9876;192.168.2.51:9876");
        // 3.启动producer
        producer.start();
        int messageCount = 10;
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < 10; i++) {
            final int index = i;
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            Message message = new Message("base", "Tag2", ("Hello RocketMQ Async Message" + (i + 1)).getBytes(Charset.defaultCharset()));
            // 5.发送异步消息
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    countDownLatch.countDown();
                    System.out.printf("%-10d OK %s %n",index,sendResult.getMsgId());
                    // System.out.println("发送结果：" + sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    countDownLatch.countDown();
                    System.out.printf("%-10d Exception %s %n",index,throwable);
                    // System.out.println("发送异常：" + throwable);
                }
            });
        }
        // 6.关闭生产者producer
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }

    // 发送同步消息
    private static void sendSyncMessage() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        // 1.创建消息生产者producer，并制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
        // 2.指定Nameserver地址
        producer.setNamesrvAddr("192.168.11.129:9876;192.168.11.130:9876");
        // producer.setNamesrvAddr("192.168.2.50:9876;192.168.2.51:9876");
        // 3.启动producer
        producer.start();
        for (int i = 0; i < 10; i++) {
            // 4.创建消息对象，指定主题Topic、Tag和消息体
            Message message = new Message("base", "Tag1", ("Hello RocketMQ" + (i + 1)).getBytes(Charset.defaultCharset()));
            // 5.发送消息
            SendResult sendResult = producer.send(message);
            SendStatus sendStatus = sendResult.getSendStatus();         // 消息发送状态
            String msgId = sendResult.getMsgId();                       // 消息ID
            int queueId = sendResult.getMessageQueue().getQueueId();    // 接收消息的队列id
            System.out.println("发送状态：" + sendStatus + ", 消息ID: " + msgId + ", 队列id: " + queueId);
            // 线程睡1秒
            TimeUnit.SECONDS.sleep(1);
        }
        // 6.关闭生产者producer
        producer.shutdown();
    }
}
